package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"sync"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		var r = NullArgs{}
		requestReplay := WorkInfo{}
		if false == call("Coordinator.RequestWork", &r, &requestReplay) {
			// fmt.Print("Call for Coordinator.RequestWork return false!")
			return
		}
		// if requestReplay.Kind == 1 && requestReplay.TaskId >= requestReplay.ReduceTaskNum {
		// 	fmt.Println("Receive Reduce Task ", requestReplay.TaskId, ", and the max reduce task should be ", requestReplay.ReduceTaskNum)
		// }
		// if requestReplay.Kind == 0 && requestReplay.TaskId >= requestReplay.MapTaskNum {
		// 	fmt.Println("Receive Map Task ", requestReplay.TaskId, ", and the max map task should be ", requestReplay.MapTaskNum)
		// }
		// fmt.Println("Coordinator.RequestWork Back ", "Kind = ", requestReplay.Kind, " status = ", requestReplay.Status, requestReplay.TaskId, "filename = ", requestReplay.Filename)
		if requestReplay.Kind == 0 {
			// fmt.Println("Kind MAP ID = ", requestReplay.TaskId, "filename = ", requestReplay.Filename)
			mapTask(mapf, requestReplay)
			// time.Sleep(1 * time.Second)
		} else if requestReplay.Kind == 1 {
			// fmt.Println("Kind REDUCE ID = ", requestReplay.TaskId, "filename = ", requestReplay.Filename)
			reduceTask(reducef, requestReplay)
			// time.Sleep(1 * time.Second)
		} else if requestReplay.Kind == -1 {
			// fmt.Println("All Work Complete!")
			return
		} else if requestReplay.Kind == 2 {
			time.Sleep(1 * time.Second)
		} else {
			fmt.Print("Replay work kind invalid, kind =", requestReplay.Kind)
			return
		}
	}

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()

}
func mapTask(mapf func(string, string) []KeyValue, requestReplay WorkInfo) {
	filename := requestReplay.Filename

	/*the following part code are copied directly from mrsequential.go,
	remove sort and append part due to one map task have only one file*/
	// intermediate := []mr.KeyValue{}
	// fmt.Println("mapTask File Get, it is ", filename)
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("mapTask cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("mapTask cannot read %v", filename)
	}
	file.Close()
	kva := mapf(filename, string(content))
	sort.Sort(ByKey(kva))

	encArray := make([]*json.Encoder, requestReplay.ReduceTaskNum)
	tempFile := make([]*os.File, requestReplay.ReduceTaskNum)
	for i := 0; i < requestReplay.ReduceTaskNum; i++ {
		tempFile[i], err = ioutil.TempFile("", "mrtemp-"+strconv.Itoa(requestReplay.TaskId)+"-"+strconv.Itoa(i)+"-*.txt")
		if err != nil {
			// fmt.Println("i = ", i, " ioutil.TempFile Failed: ", err)
			break
		}
		encArray[i] = json.NewEncoder(tempFile[i])
		// fmt.Println("Map File ", i, " Read Complete")
	}

	for _, kv := range kva {
		err := encArray[ihash(kv.Key)%(requestReplay.ReduceTaskNum)].Encode(kv)
		if err != nil {
			break
		}
	}

	for i, f := range tempFile {
		f.Close()
		writeFile(f.Name(), "mr-"+strconv.Itoa(requestReplay.TaskId)+"-"+strconv.Itoa(i))
		// os.Rename(f.Name(), "mr-"+strconv.Itoa(requestReplay.TaskId)+"-"+strconv.Itoa(i))
	}
	n := NullArgs{}
	call("Coordinator.CompleteTaskNotify", &requestReplay, &n)

}

func reduceTask(reducef func(string, []string) string, requestReplay WorkInfo) {
	intermediate := []KeyValue{}
	// fmt.Println("Receive Reduce TaskID = ", requestReplay.TaskId, " and total Map Num is ", requestReplay.MapTaskNum)
	for i := 0; i < requestReplay.MapTaskNum; i++ {
		filename := "mr-" + strconv.Itoa(i) + "-" + strconv.Itoa(requestReplay.TaskId)
		file, err := os.Open(filename)
		if err != nil {
			// fmt.Println("Reduce ", i, " ReadFile Failed! filename is ", filename, " err is", err)
			continue
		} else {
			// fmt.Println("Reduce ", i, " ReadFile Success! filename is ", filename)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				if err != io.EOF {
					fmt.Println("Decode err = ", err)
				}
				break
			}
			intermediate = append(intermediate, kv)
		}
		file.Close()
		// os.Remove(filename)
	}
	sort.Sort(ByKey(intermediate))
	/*the following part code are copied directly from mrsequential.go*/
	ofile, _ := ioutil.TempFile("", "mr-out-t"+strconv.Itoa(requestReplay.TaskId))
	// ofile, _ := os.Create(otname)

	//
	// call Reduce on each distinct key in intermediate[],
	// and print the result to mr-out-0.
	//
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}

	ofile.Close()
	n := NullArgs{}
	// os.Rename(ofile.Name(), "mr-out-t"+strconv.Itoa(requestReplay.TaskId))
	writeFile(ofile.Name(), "mr-out-t"+strconv.Itoa(requestReplay.TaskId))
	call("Coordinator.CompleteTaskNotify", &requestReplay, &n)
}

var filemux sync.Mutex

func writeFile(oldname, newname string) {
	/*
		Fix os.Rename caused problems
		REF: https://golang.google.cn/pkg/os/#Rename
		func Rename
		func Rename(oldpath, newpath string) error
		Rename renames (moves) oldpath to newpath.
		If newpath already exists and is not a directory, Rename replaces it.
		OS-specific restrictions may apply when oldpath and newpath are in different directories.
		If there is an error, it will be of type *LinkError.
	*/
	filemux.Lock()
	defer filemux.Unlock()
	f, ok := os.Open(newname)
	defer f.Close()
	if ok == nil {
		/*File exist, return directly*/
		return
	}
	os.Rename(oldname, newname)
}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
